// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2024-12-05

package scheduler_http

import (
	"context"
	"fmt"
	"gitee.com/go-libs/log"
	"gitee.com/gomq/sr"
	"github.com/valyala/fasthttp"
	"net/http"
	"sync"
	"time"
)

var dispatcherPool = &sync.Pool{
	New: func() any {
		return (&dispatcher{}).acquirer()
	},
}

type dispatcher struct {
	endpoint sr.Endpoint
	message  *sr.Message
	request  *fasthttp.Request
	response *fasthttp.Response
}

// Builder acquire a sr.Scheduler instance for http or https from pool, You must
// call Release() method to put into pool when scheduler end.
func Builder() sr.Scheduler {
	return dispatcherPool.
		Get().(*dispatcher).
		acquirer()
}

// Dispatch
// message to given endpoint by http or https request.
func (o *dispatcher) Dispatch(ctx context.Context, endpoint sr.Endpoint, message *sr.Message, data []byte) (response sr.Response) {
	log.Infofc(ctx, `[scheduler=http] schedule begin: request="HTTP %s %s", format="%s"`, endpoint.GetMethod(), endpoint.GetAddress(), endpoint.GetFormat())

	// Bind instance references.
	o.endpoint, o.message = endpoint, message

	// Prepare dispatch response.
	response = sr.NewResponse()

	// Triggered when process done.
	defer func() {
		// Recover and catch runtime panic if necessary.
		if r := recover(); r != nil {
			// Restore error reason.
			err := sr.NewError(http.StatusInternalServerError, fmt.Sprintf(`%v`, r))
			response.SetError(err)

			log.Fatalfc(ctx, `[scheduler=http] panic on schedule: %s`, err)
		}

		// End dispatch.
		response.End()

		if response.HasError() {
			log.Warnfc(ctx, `[scheduler=http] schedule failed: duration="%dms", error="%v"`, response.GetDuration().Milliseconds(), response.GetError())
		} else {
			log.Infofc(ctx, `[scheduler=http] schedule succeed: duration="%dms"`, response.GetDuration().Milliseconds())
		}
	}()

	// scheduler child processes.
	o.doBasic()
	o.doTracing(ctx)
	o.doHeader()
	o.doData(data)
	o.doSend(response)
	return
}

// Release
// put sr.Scheduler instance into pool.
func (o *dispatcher) Release() {
	o.releaser()
}

// + Access

func (o *dispatcher) acquirer() *dispatcher {
	o.request = fasthttp.AcquireRequest()
	o.response = fasthttp.AcquireResponse()
	return o
}

func (o *dispatcher) doBasic() {
	o.request.SetRequestURI(o.endpoint.GetAddress().String())
	o.request.Header.SetMethod(o.endpoint.GetMethod().String())
}

func (o *dispatcher) doData(data []byte) {
	// No message body.
	if o.message.Body == nil {
		return
	}

	// Configure
	// request content type.
	o.request.Header.SetContentType(o.endpoint.GetFormat().ContentType())

	// JSON or XML.
	if o.endpoint.GetFormat().IsJson() || o.endpoint.GetFormat().IsXml() {
		o.request.SetBody(data)
		return
	}

	// Normal message body.
	o.request.SetBody(data)
}

func (o *dispatcher) doHeader() {
	for k, v := range map[string]any{
		"Mq-Message-Consumed-Count":     o.message.ConsumedCount,
		"Mq-Message-Delay-Milliseconds": o.message.Delay.Milliseconds(),
		"Mq-Message-Id":                 o.message.MessageId,
		"Mq-Message-Queue":              o.message.Queue,
		"Mq-Message-Tag":                o.message.Tag,
		"Mq-Message-Time":               o.message.Time.Format(time.RFC3339Nano),
		"Mq-Message-Topic":              o.message.Topic,
	} {
		o.request.Header.Set(k, fmt.Sprintf(`%v`, v))
	}
}

func (o *dispatcher) doSend(response sr.Response) {
	// Send request.
	if err := fasthttp.DoTimeout(o.request, o.response, o.endpoint.GetTimeout()); err != nil {
		response.SetError(err)
		return
	}

	// Bind response body.
	response.SetBody(o.response.Body())

	// Verify status code.
	if code := o.response.StatusCode(); code != http.StatusOK {
		response.SetError(sr.NewError(code, http.StatusText(code)))
	}
}

func (o *dispatcher) doTracing(ctx context.Context) {
	// Iterate
	// opentracing fields.
	//
	//   - X-B3-Traceid
	//   - X-B3-Spanid
	//   - X-B3-SpanPosition
	for k, vs := range log.ContextHeader(ctx) {
		if len(vs) > 0 && vs[0] != "" {
			o.request.Header.Set(k, vs[0])
		}
	}
}

func (o *dispatcher) releaser() {
	fasthttp.ReleaseRequest(o.request)
	fasthttp.ReleaseResponse(o.response)

	o.endpoint, o.message, o.request, o.response = nil, nil, nil, nil
	dispatcherPool.Put(o)
}
